package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentInfo;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class PaymentWideApp {
    public static void main(String[] args)  throws  Exception{
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        //TODO 1.接收数据流
        String groupId = "payment_wide_group";
        String paymentInfoSourceTopic ="dwd_payment_info";
        String orderWideSourceTopic ="dwm_order_wide";
        String paymentWideSinkTopic = "dwm_payment_wide";

        //封装kafka消费者，读取支付数据
        FlinkKafkaConsumer<String> paymentInfoSource = MyKafkaUtil.getKafkaSource(paymentInfoSourceTopic,groupId);
        DataStream<String> paymentInfojsonDStream = env.addSource(paymentInfoSource);
        DataStream<PaymentInfo> paymentInfoDataStream =
                paymentInfojsonDStream.map(jsonString -> JSON.parseObject(jsonString,PaymentInfo.class));

        //封装kafka消费者，订单宽表数据
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic,groupId);
        DataStream<String> orderWidejsonDStream = env.addSource(orderWideSource);
        DataStream<OrderWide> orderWideDStream = orderWidejsonDStream.map(jsonString -> JSON.parseObject(jsonString,OrderWide.class));

        //TODO 1.设置水位线
        SingleOutputStreamOperator<PaymentInfo> paymentWideSingleOutputStreamOperator =
                paymentInfoDataStream.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((paymentInfo,ts) ->
                                DateTimeUtil.toTs(paymentInfo.getCallback_time()))
                );

        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperator = orderWideDStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (SerializableTimestampAssigner<OrderWide>) (orderWide, l) -> DateTimeUtil.toTs(orderWide.getCreate_time())
                )
        );


        //TODO 2. 设置分区键
        KeyedStream<PaymentInfo,Long> paymentInfoLongKeyedStream = paymentWideSingleOutputStreamOperator.keyBy(PaymentInfo::getOrder_id);
        KeyedStream<OrderWide,Long> orderWideLongKeyedStream = orderWideSingleOutputStreamOperator.keyBy(OrderWide::getOrder_id);

        //TODO 3.关联数据
        SingleOutputStreamOperator<PaymentWide> paymentWideDStream = paymentInfoLongKeyedStream.intervalJoin(
                orderWideLongKeyedStream)
                .between(Time.seconds(-1800),Time.seconds(0))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, Context context, Collector<PaymentWide> collector) throws Exception {
                        collector.collect(new PaymentWide(paymentInfo,orderWide));
                    }
                }).uid("payment_wide_join");

        // TODO 4. 输出
        SingleOutputStreamOperator<String> paymentWideStringDStream = paymentWideDStream.map(JSON::toJSONString);
        paymentWideStringDStream.print("pay:::");

        paymentWideStringDStream.addSink(
                MyKafkaUtil.getKafkaSink(paymentWideSinkTopic)
        );

        env.execute();





    }
}
